/**
 * Captain
 * © 2017 Captain copyright，All rights reserved.
 * http://www.sccaptain.com.cn
 * 
 * JAVA : 8
 * 文  件  名: BaseJpaDao.java
 * 创  建  人: Roc Zheng (roc.djz@gmail.com)
 * 创建时间: 2017年6月28日 上午8:59:01
 * 版         本: 1.0.0
 * 备         注:
 * 修订历史:
 */
package com.material.c2c.core.dao.datajpa;

import static org.springframework.data.jpa.repository.query.QueryUtils.COUNT_QUERY_STRING;
import static org.springframework.data.jpa.repository.query.QueryUtils.DELETE_ALL_QUERY_STRING;
import static org.springframework.data.jpa.repository.query.QueryUtils.applyAndBind;
import static org.springframework.data.jpa.repository.query.QueryUtils.getQueryString;
import static org.springframework.data.jpa.repository.query.QueryUtils.toOrders;

import java.io.Serializable;
import java.lang.reflect.ParameterizedType;
import java.lang.reflect.Type;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import javax.persistence.EntityManager;
import javax.persistence.LockModeType;
import javax.persistence.NoResultException;
import javax.persistence.Parameter;
import javax.persistence.PersistenceContext;
import javax.persistence.Query;
import javax.persistence.TypedQuery;
import javax.persistence.criteria.CriteriaBuilder;
import javax.persistence.criteria.CriteriaQuery;
import javax.persistence.criteria.Order;
import javax.persistence.criteria.ParameterExpression;
import javax.persistence.criteria.Path;
import javax.persistence.criteria.Predicate;
import javax.persistence.criteria.Root;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.data.domain.Example;
import org.springframework.data.domain.Page;
import org.springframework.data.domain.PageImpl;
import org.springframework.data.domain.Pageable;
import org.springframework.data.domain.Sort;
import org.springframework.data.domain.Sort.Direction;
import org.springframework.data.jpa.convert.QueryByExamplePredicateBuilder;
import org.springframework.data.jpa.domain.Specification;
import org.springframework.data.jpa.provider.PersistenceProvider;
import org.springframework.data.jpa.repository.EntityGraph;
import org.springframework.data.jpa.repository.JpaRepository;
import org.springframework.data.jpa.repository.JpaSpecificationExecutor;
import org.springframework.data.jpa.repository.query.Jpa21Utils;
import org.springframework.data.jpa.repository.query.JpaEntityGraph;
import org.springframework.data.jpa.repository.query.QueryUtils;
import org.springframework.data.jpa.repository.support.CrudMethodMetadata;
import org.springframework.data.jpa.repository.support.JpaEntityInformation;
import org.springframework.data.jpa.repository.support.JpaEntityInformationSupport;
import org.springframework.data.jpa.repository.support.SimpleJpaRepository;
import org.springframework.data.repository.support.PageableExecutionUtils;
import org.springframework.data.repository.support.PageableExecutionUtils.TotalSupplier;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;

import com.material.c2c.core.dao.EntityDao;

/**
 * 
 * 
 * @since 1.0
 * @version 1.0
 * @author Roc Zheng (roc.djz@gmail.com)
 */
public class BaseEntityJpaDao<T, ID extends Serializable> implements EntityDao<T, ID>, JpaRepository<T, ID>, JpaSpecificationExecutor<T> {
  private static final String ID_MUST_NOT_BE_NULL = "id 不能为空。";

  protected Class<T> entityClass;
  protected final Logger logger = LoggerFactory.getLogger(getClass());

  private EntityManager entityManager;
  private CrudMethodMetadata metadata;
  private PersistenceProvider provider;
  private JpaEntityInformation<T, ?> entityInformation;

  /**
   * 获取 entityManager。
   * 
   * @return entityManager。
   */
  public EntityManager getEntityManager() {
    return entityManager;
  }

  /**
   * @param entityManager 要设置的 entityManager。
   */
  @PersistenceContext
  public void setEntityManager(final EntityManager entityManager) {
    Assert.notNull(entityManager, "entityManager 不能为空。");
    this.entityManager = entityManager;
    this.provider = PersistenceProvider.fromEntityManager(entityManager);
    this.entityInformation = JpaEntityInformationSupport.getEntityInformation(entityClass, entityManager);
  }

  /**
   * Configures a custom {@link CrudMethodMetadata} to be used to detect
   * {@link LockModeType}s and query hints to be applied to queries.
   * 
   * @param crudMethodMetadata
   */
  public void setMethodMetadata(CrudMethodMetadata crudMethodMetadata) {
    this.metadata = crudMethodMetadata;
  }

  protected CrudMethodMetadata getMethodMetadata() {
    return metadata;
  }

  protected Class<T> getEntityClass() {
    return entityInformation.getJavaType();
  }

  private String getDeleteAllQueryString() {
    return getQueryString(DELETE_ALL_QUERY_STRING, entityInformation.getEntityName());
  }

  private String getCountQueryString() {
    String countQuery = String.format(COUNT_QUERY_STRING, provider.getCountQueryPlaceholder(), "%s");
    return getQueryString(countQuery, entityInformation.getEntityName());
  }

  /**
   * 初始化 {@link BaseEntityJpaDao} 类的新实例。
   */
  @SuppressWarnings("unchecked")
  public BaseEntityJpaDao() {
    Type genType = getClass().getGenericSuperclass();
    if (genType instanceof ParameterizedType) {
      Type[] params = ((ParameterizedType) genType).getActualTypeArguments();
      entityClass = (Class<T>) params[0];
    }
  }
  
  /**
   * 初始化 {@link BaseEntityJpaDao} 类的新实例。
   * @param entityClass
   */
  public BaseEntityJpaDao(Class<T> entityClass) {
    this.entityClass = entityClass;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  @Transactional
  public <S extends T> S add(S entity) {
    entityManager.persist(entity);
    return entity;
  }

  /**
   * {@inheritDoc}
   */
  @Override
  @Transactional
  public <S extends T> S update(S entity) {
    return entityManager.merge(entity);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  @Transactional
  public void delete(ID id) {
    Assert.notNull(id, ID_MUST_NOT_BE_NULL);
    T entity = findOne(id);

    if (entity == null) {
      logger.error(String.format("不存在 ID 为  %s 的表实体 %s。", entityInformation.getJavaType(), id));
      // throw new EmptyResultDataAccessException(String.format("No %s entity
      // with id %s exists!", entityInformation.getJavaType(), id), 1);
      return;
    }

    delete(entity);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  @Transactional
  public void delete(T entity) {
    Assert.notNull(entity, "entity 不能为空。");
    entityManager.remove(entityManager.contains(entity) ? entity : entityManager.merge(entity));
  }

  /**
   * {@inheritDoc}
   */
  @Transactional
  public void delete(Iterable<? extends T> entities) {
    Assert.notNull(entities, "entities 不能为空。");

    for (T entity : entities) {
      delete(entity);
    }
  }

  /**
   * {@inheritDoc}
   */
  @Transactional
  public void deleteInBatch(Iterable<T> entities) {
    Assert.notNull(entities, "entities 不能为空。");

    if (!entities.iterator().hasNext()) {
      return;
    }

    applyAndBind(getQueryString(DELETE_ALL_QUERY_STRING, entityInformation.getEntityName()), entities, entityManager)
        .executeUpdate();
  }

  /**
   * {@inheritDoc}
   */
  @Transactional
  public void deleteAll() {
    for (T element : findAll()) {
      delete(element);
    }
  }

  /**
   * {@inheritDoc}
   */
  @Transactional
  public void deleteAllInBatch() {
    entityManager.createQuery(getDeleteAllQueryString()).executeUpdate();
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public T findOne(ID id) {
    Assert.notNull(id, ID_MUST_NOT_BE_NULL);

    Class<T> domainType = getEntityClass();

    if (metadata == null) {
      return entityManager.find(domainType, id);
    }

    LockModeType type = metadata.getLockModeType();
    Map<String, Object> hints = getQueryHints();
    return type == null ? entityManager.find(domainType, id, hints) : entityManager.find(domainType, id, type, hints);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public T getOne(ID id) {
    Assert.notNull(id, ID_MUST_NOT_BE_NULL);
    return entityManager.getReference(getEntityClass(), id);
  }

  /**
   * {@inheritDoc}
   */
  @Override
  public List<T> listAll() {
    return findAll();
  }

  /**
   * 
   * @param ids
   * @return
   * 
   * @since 1.0
   * @version 1.0
   * @author Roc Zheng (roc.djz@gmail.com)
   */
  public List<T> listAll(Iterable<ID> ids) {
    return findAll(ids);
  }

  /**
   * 
   * @param sort
   * @return
   * 
   * @since 1.0
   * @version 1.0
   * @author Roc Zheng (roc.djz@gmail.com)
   */
  public List<T> listAll(Sort sort) {
    return findAll(sort);
  }

  /**
   * 
   * @param specification
   * @return
   * 
   * @since 1.0
   * @version 1.0
   * @author Roc Zheng (roc.djz@gmail.com)
   */
  public List<T> listAll(Specification<T> specification) {
    return findAll(specification);
  }

  /**
   * 
   * @param example
   * @return
   * 
   * @since 1.0
   * @version 1.0
   * @author Roc Zheng (roc.djz@gmail.com)
   */
  public <S extends T> List<S> listAll(Example<S> example) {
    return findAll(example);
  }

  /**
   * 
   * @param specification
   * @param sort
   * @return
   * 
   * @since 1.0
   * @version 1.0
   * @author Roc Zheng (roc.djz@gmail.com)
   */
  public List<T> listAll(Specification<T> specification, Sort sort) {
    return findAll(specification, sort);
  }

  /**
   * 
   * @param example
   * @param sort
   * @return
   * 
   * @since 1.0
   * @version 1.0
   * @author Roc Zheng (roc.djz@gmail.com)
   */
  public <S extends T> List<S> listAll(Example<S> example, Sort sort) {
    return findAll(example, sort);
  }

  /**
   * 
   * @param pageable
   * @return
   * 
   * @since 1.0
   * @version 1.0
   * @author Roc Zheng (roc.djz@gmail.com)
   */
  public Page<T> pageAll(Pageable pageable) {
    return findAll(pageable);
  }

  /**
   * 
   * @param specification
   * @param pageable
   * @return
   * 
   * @since 1.0
   * @version 1.0
   * @author Roc Zheng (roc.djz@gmail.com)
   */
  public Page<T> pageAll(Specification<T> specification, Pageable pageable) {
    return findAll(specification, pageable);
  }

  /**
   * 
   * @param example
   * @param pageable
   * @return
   * 
   * @since 1.0
   * @version 1.0
   * @author Roc Zheng (roc.djz@gmail.com)
   */
  public <S extends T> Page<S> pageAll(Example<S> example, Pageable pageable) {
    return findAll(example, pageable);
  }

  /**
   * {@inheritDoc}
   */
  public boolean exists(ID id) {
    Assert.notNull(id, ID_MUST_NOT_BE_NULL);

    if (entityInformation.getIdAttribute() == null) {
      return findOne(id) != null;
    }

    String placeholder = provider.getCountQueryPlaceholder();
    String entityName = entityInformation.getEntityName();
    Iterable<String> idAttributeNames = entityInformation.getIdAttributeNames();
    String existsQuery = QueryUtils.getExistsQueryString(entityName, placeholder, idAttributeNames);

    TypedQuery<Long> query = entityManager.createQuery(existsQuery, Long.class);

    if (!entityInformation.hasCompositeId()) {
      query.setParameter(idAttributeNames.iterator().next(), id);
      return query.getSingleResult() == 1L;
    }

    for (String idAttributeName : idAttributeNames) {
      Object idAttributeValue = entityInformation.getCompositeIdAttributeValue(id, idAttributeName);
      boolean complexIdParameterValueDiscovered = idAttributeValue != null
          && !query.getParameter(idAttributeName).getParameterType().isAssignableFrom(idAttributeValue.getClass());

      if (complexIdParameterValueDiscovered) {
        // fall-back to findOne(id) which does the proper mapping for the parameter.
        return findOne(id) != null;
      }

      query.setParameter(idAttributeName, idAttributeValue);
    }

    return query.getSingleResult() == 1L;
  }

  /*
   * (non-Javadoc)
   * 
   * @see org.springframework.data.jpa.repository.JpaRepository#findAll()
   */
  public List<T> findAll() {
    return getQuery(null, (Sort) null).getResultList();
  }

  /*
   * (non-Javadoc)
   * 
   * @see org.springframework.data.repository.CrudRepository#findAll(ID[])
   */
  public List<T> findAll(Iterable<ID> ids) {
    if (ids == null || !ids.iterator().hasNext()) {
      return Collections.emptyList();
    }

    if (entityInformation.hasCompositeId()) {
      List<T> results = new ArrayList<T>();
      for (ID id : ids) {
        results.add(findOne(id));
      }

      return results;
    }

    ByIdsSpecification<T> specification = new ByIdsSpecification<T>(entityInformation);
    TypedQuery<T> query = getQuery(specification, (Sort) null);

    return query.setParameter(specification.parameter, ids).getResultList();
  }

  /*
   * (non-Javadoc)
   * 
   * @see org.springframework.data.jpa.repository.JpaRepository#findAll(org.
   * springframework.data.domain.Sort)
   */
  public List<T> findAll(Sort sort) {
    return getQuery(null, sort).getResultList();
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.repository.PagingAndSortingRepository#findAll(org.
   * springframework.data.domain.Pageable)
   */
  public Page<T> findAll(Pageable pageable) {

    if (null == pageable) {
      return new PageImpl<T>(findAll());
    }

    return findAll((Specification<T>) null, pageable);
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.jpa.repository.JpaSpecificationExecutor#findOne(
   * org.springframework.data.jpa.domain.Specification)
   */
  public T findOne(Specification<T> spec) {
    try {
      return getQuery(spec, (Sort) null).getSingleResult();
    } catch (NoResultException e) {
      return null;
    }
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.jpa.repository.JpaSpecificationExecutor#findAll(
   * org.springframework.data.jpa.domain.Specification)
   */
  public List<T> findAll(Specification<T> spec) {
    return getQuery(spec, (Sort) null).getResultList();
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.jpa.repository.JpaSpecificationExecutor#findAll(
   * org.springframework.data.jpa.domain.Specification,
   * org.springframework.data.domain.Pageable)
   */
  public Page<T> findAll(Specification<T> spec, Pageable pageable) {
    TypedQuery<T> query = getQuery(spec, pageable);
    return pageable == null ? new PageImpl<T>(query.getResultList())
        : readPage(query, getEntityClass(), pageable, spec);
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.jpa.repository.JpaSpecificationExecutor#findAll(
   * org.springframework.data.jpa.domain.Specification,
   * org.springframework.data.domain.Sort)
   */
  @Override
  public List<T> findAll(Specification<T> spec, Sort sort) {
    return getQuery(spec, sort).getResultList();
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.repository.query.QueryByExampleExecutor#findOne(
   * org.springframework.data.domain.Example)
   */
  @Override
  public <S extends T> S findOne(Example<S> example) {
    try {
      return getQuery(new ExampleSpecification<S>(example), example.getProbeType(), (Sort) null).getSingleResult();
    } catch (NoResultException e) {
      return null;
    }
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.repository.query.QueryByExampleExecutor#count(org.
   * springframework.data.domain.Example)
   */
  @Override
  public <S extends T> long count(Example<S> example) {
    return executeCountQuery(getCountQuery(new ExampleSpecification<S>(example), example.getProbeType()));
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.repository.query.QueryByExampleExecutor#exists(org
   * .springframework.data.domain.Example)
   */
  @Override
  public <S extends T> boolean exists(Example<S> example) {
    return !getQuery(new ExampleSpecification<S>(example), example.getProbeType(), (Sort) null).getResultList()
        .isEmpty();
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.repository.query.QueryByExampleExecutor#findAll(
   * org.springframework.data.domain.Example)
   */
  @Override
  public <S extends T> List<S> findAll(Example<S> example) {
    return getQuery(new ExampleSpecification<S>(example), example.getProbeType(), (Sort) null).getResultList();
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.repository.query.QueryByExampleExecutor#findAll(
   * org.springframework.data.domain.Example,
   * org.springframework.data.domain.Sort)
   */
  @Override
  public <S extends T> List<S> findAll(Example<S> example, Sort sort) {
    return getQuery(new ExampleSpecification<S>(example), example.getProbeType(), sort).getResultList();
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.repository.query.QueryByExampleExecutor#findAll(
   * org.springframework.data.domain.Example,
   * org.springframework.data.domain.Pageable)
   */
  @Override
  public <S extends T> Page<S> findAll(Example<S> example, Pageable pageable) {
    ExampleSpecification<S> spec = new ExampleSpecification<S>(example);
    Class<S> probeType = example.getProbeType();
    TypedQuery<S> query = getQuery(new ExampleSpecification<S>(example), probeType, pageable);

    return pageable == null ? new PageImpl<S>(query.getResultList()) : readPage(query, probeType, pageable, spec);
  }

  /*
   * (non-Javadoc)
   * 
   * @see org.springframework.data.repository.CrudRepository#count()
   */
  @Override
  public long count() {
    return entityManager.createQuery(getCountQueryString(), Long.class).getSingleResult();
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.jpa.repository.JpaSpecificationExecutor#count(org.
   * springframework.data.jpa.domain.Specification)
   */
  public long count(Specification<T> spec) {
    return executeCountQuery(getCountQuery(spec, getEntityClass()));
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.repository.CrudRepository#save(java.lang.Object)
   */
  @Transactional
  public <S extends T> S save(S entity) {
    if (entityInformation.isNew(entity)) {
      entityManager.persist(entity);
      return entity;
    } else {
      return entityManager.merge(entity);
    }
  }

  /*
   * (non-Javadoc)
   * 
   * @see
   * org.springframework.data.jpa.repository.JpaRepository#saveAndFlush(java.
   * lang.Object)
   */
  @Transactional
  public <S extends T> S saveAndFlush(S entity) {
    S result = save(entity);
    flush();
    return result;
  }

  /*
   * (non-Javadoc)
   * 
   * @see org.springframework.data.jpa.repository.JpaRepository#save(java.lang.
   * Iterable)
   */
  @Transactional
  public <S extends T> List<S> save(Iterable<S> entities) {
    List<S> result = new ArrayList<S>();
    if (entities == null) {
      return result;
    }

    for (S entity : entities) {
      result.add(save(entity));
    }

    return result;
  }

  /*
   * (non-Javadoc)
   * 
   * @see org.springframework.data.jpa.repository.JpaRepository#flush()
   */
  @Transactional
  public void flush() {
    entityManager.flush();
  }
  
  public Page<T> pageByJpql(String jpql, Pageable pageable, Map<String, Object> params) {
    return pageByJpql(jpql, null, null, null, getEntityClass(), pageable, params);
  }
  
  public <S> Page<S> pageByJpql(String jpql, Class<S> resultClass, Pageable pageable, Map<String, Object> params) {
    return pageByJpql(jpql, null, null, null, resultClass, pageable, params);
  }
  
  public Page<T> pageByJpql(String jpql, String orderEntityAlias, Pageable pageable, Map<String, Object> params) {
    return pageByJpql(jpql, null, null, orderEntityAlias, getEntityClass(), pageable, params);
  }
  
  public <S> Page<S> pageByJpql(String jpql, String orderEntityAlias, Class<S> resultClass, Pageable pageable, Map<String, Object> params) {
    return pageByJpql(jpql, null, null, orderEntityAlias, resultClass, pageable, params);
  }
  
  public Page<T> pageByJpql(String jpql, String countDistinctProperty, String countDistinctEntityAlias, String orderEntityAlias, Pageable pageable, Map<String, Object> params) {
    return pageByJpql(jpql, countDistinctProperty, countDistinctEntityAlias, orderEntityAlias, getEntityClass(), pageable, params);
  }
  
  public <S> Page<S> pageByJpql(String jpql, String countDistinctProperty, String countDistinctEntityAlias, String orderEntityAlias, Class<S> resultClass, Pageable pageable, Map<String, Object> params) {
    TypedQuery<Long> countQuery = getCountQuery(jpql, countDistinctProperty, countDistinctEntityAlias);
    for (Entry<String, Object> entry : params.entrySet()) {
      countQuery.setParameter(entry.getKey(), entry.getValue());
    }
    long total = executeCountQuery(countQuery);
    
    TypedQuery<S> dataQuery = getQuery(jpql, resultClass, pageable, orderEntityAlias);
    for (Entry<String, Object> entry : params.entrySet()) {
      dataQuery.setParameter(entry.getKey(), entry.getValue());
    }
    dataQuery.setFirstResult(pageable.getOffset()).setMaxResults(pageable.getPageSize());
    
    return new PageImpl<>(dataQuery.getResultList(), pageable, total);
  }
  
  public Page<T> pageByJpql(String jpql, Pageable pageable, Object... params) {
    return pageByJpql(jpql, null, null, null, getEntityClass(), pageable, Arrays.asList(params));
  }
  
  public <S> Page<S> pageByJpql(String jpql, Class<S> resultClass, Pageable pageable, Object... params) {
    return pageByJpql(jpql, null, null, null, resultClass, pageable, Arrays.asList(params));
  }
  
  public Page<T> pageByJpql(String jpql, String orderEntityAlias, Pageable pageable, Object... params) {
    return pageByJpql(jpql, null, null, orderEntityAlias, getEntityClass(), pageable, Arrays.asList(params));
  }
  
  public <S> Page<S> pageByJpql(String jpql, String orderEntityAlias, Class<S> resultClass, Pageable pageable, Object... params) {
    return pageByJpql(jpql, null, null, orderEntityAlias, resultClass, pageable, Arrays.asList(params));
  }
  
  public Page<T> pageByJpql(String jpql, String orderEntityAlias, Pageable pageable, Iterable<Object> params) {
    return pageByJpql(jpql, null, null, orderEntityAlias, getEntityClass(), pageable, params);
  }
  
  public <S> Page<S> pageByJpql(String jpql, String orderEntityAlias, Class<S> resultClass, Pageable pageable, Iterable<Object> params) {
    return pageByJpql(jpql, null, null, orderEntityAlias, resultClass, pageable, params);
  }
  
  public <S> Page<S> pageByJpql(String jpql, String countDistinctProperty, String countDistinctEntityAlias, String orderEntityAlias, Class<S> resultClass, Pageable pageable, Iterable<Object> params) {
    TypedQuery<Long> countQuery = getCountQuery(jpql, countDistinctProperty, countDistinctEntityAlias);
    int paramIndex = 0;
    for (Object param : params) {
      countQuery.setParameter(paramIndex++, param);
    }
    long total = executeCountQuery(countQuery);
    
    TypedQuery<S> dataQuery = getQuery(jpql, resultClass, pageable, orderEntityAlias);
    paramIndex = 0;
    for (Object param : params) {
      dataQuery.setParameter(paramIndex++, param);
    }
    dataQuery.setFirstResult(pageable.getOffset()).setMaxResults(pageable.getPageSize());
    
    return new PageImpl<S>(dataQuery.getResultList(), pageable, total);
  }
  
  protected <S> TypedQuery<S> getQuery(String jpql, Class<S> resultClass, Pageable pageable, String entityAlias) {
    Sort sort = pageable == null ? null : pageable.getSort();
    jpql = applySortToJqpl(jpql, sort, entityAlias);
    TypedQuery<S> dataQuery = entityManager.createQuery(jpql, resultClass);
    return dataQuery;
  }
  
  protected TypedQuery<Long> getCountQuery(String jpql) {
    return getCountQuery(jpql, null, null);
  }
  
  protected TypedQuery<Long> getCountQuery(String jpql, String distinctProperty) {
    return getCountQuery(jpql, distinctProperty, null);
  }
  
  protected TypedQuery<Long> getCountQuery(String jpql, String distinctProperty, String distinctEntityAlias) {
    StringBuilder sb = new StringBuilder("select count(");
    if (StringUtils.hasText(distinctProperty)) {
      sb.append("distinct ");
      if (StringUtils.hasText(distinctEntityAlias)) {
        sb.append(distinctEntityAlias).append(".");
      }
      sb.append(distinctProperty);
    } else {
      sb.append("1");
    }
    
    sb.append(") ").append(jpql.substring(jpql.toLowerCase().indexOf("from")));
    return entityManager.createQuery(sb.toString(), Long.class);
  }
  
  protected String applyStartsWith(String value) {
    return value + "%";
  }
  
  protected String applyContains(String value) {
    return "%" + value + "%";  
  }
  
  protected String applyEndsWith(String value) {
    return "%" + value;
  }
  
  private static String applySortToJqpl(String jpql, Sort sort, String entityAlias) {
    if (null != sort) {
      StringBuilder sb = new StringBuilder(jpql).append(" order by");
      boolean first = true;
      for (Sort.Order order : sort) {
        if (!first) {
          sb.append(",");
        }
        
        sb.append(" ");
        if (StringUtils.hasText(entityAlias)) {
          sb.append(entityAlias).append(".");
        }
        sb.append(order.getProperty());
        if (Direction.DESC.equals(order.getDirection())) {
          sb.append(" desc");
        }
      }
      jpql = sb.toString();
    }
    
    return jpql;
  }

  /**
   * Returns a {@link Map} with the query hints based on the current
   * {@link CrudMethodMetadata} and potential {@link EntityGraph} information.
   * 
   * @return
   */
  protected Map<String, Object> getQueryHints() {
    if (metadata.getEntityGraph() == null) {
      return metadata.getQueryHints();
    }

    Map<String, Object> hints = new HashMap<String, Object>();
    hints.putAll(metadata.getQueryHints());
    hints.putAll(Jpa21Utils.tryGetFetchGraphHints(entityManager, getEntityGraph(), getEntityClass()));

    return hints;
  }

  private JpaEntityGraph getEntityGraph() {
    String fallbackName = this.entityInformation.getEntityName() + "." + metadata.getMethod().getName();
    return new JpaEntityGraph(metadata.getEntityGraph(), fallbackName);
  }

  /**
   * Reads the given {@link TypedQuery} into a {@link Page} applying the given
   * {@link Pageable} and {@link Specification}.
   *
   * @param query  must not be {@literal null}.
   * @param spec can be {@literal null}.
   * @param pageable can be {@literal null}.
   * @return
   */
  protected Page<T> readPage(TypedQuery<T> query, Pageable pageable, Specification<T> spec) {
    return readPage(query, getEntityClass(), pageable, spec);
  }

  /**
   * Reads the given {@link TypedQuery} into a {@link Page} applying the given
   * {@link Pageable} and {@link Specification}.
   *
   * @param query must not be {@literal null}.
   * @param entityClass must not be {@literal null}.
   * @param spec can be {@literal null}.
   * @param pageable can be {@literal null}.
   * @return
   */
  protected <S extends T> Page<S> readPage(TypedQuery<S> query, final Class<S> entityClass, Pageable pageable, final Specification<S> spec) {
    query.setFirstResult(pageable.getOffset());
    query.setMaxResults(pageable.getPageSize());

    return PageableExecutionUtils.getPage(query.getResultList(), pageable, new TotalSupplier() {
      @Override
      public long get() {
        return executeCountQuery(getCountQuery(spec, entityClass));
      }
    });
  }

  /**
   * Creates a new {@link TypedQuery} from the given {@link Specification}.
   *
   * @param spec can be {@literal null}.
   * @param pageable can be {@literal null}.
   * @return
   */
  protected TypedQuery<T> getQuery(Specification<T> spec, Pageable pageable) {
    Sort sort = pageable == null ? null : pageable.getSort();
    return getQuery(spec, getEntityClass(), sort);
  }

  /**
   * Creates a new {@link TypedQuery} from the given {@link Specification}.
   *
   * @param spec can be {@literal null}.
   * @param entityClass must not be {@literal null}.
   * @param pageable can be {@literal null}.
   * @return
   */
  protected <S extends T> TypedQuery<S> getQuery(Specification<S> spec, Class<S> entityClass, Pageable pageable) {
    Sort sort = pageable == null ? null : pageable.getSort();
    return getQuery(spec, entityClass, sort);
  }

  /**
   * Creates a {@link TypedQuery} for the given {@link Specification} and {@link Sort}.
   * 
   * @param spec can be {@literal null}.
   * @param sort can be {@literal null}.
   * @return
   */
  protected TypedQuery<T> getQuery(Specification<T> spec, Sort sort) {
    return getQuery(spec, getEntityClass(), sort);
  }

  /**
   * Creates a {@link TypedQuery} for the given {@link Specification} and {@link Sort}.
   *
   * @param spec can be {@literal null}.
   * @param entityClass must not be {@literal null}.
   * @param sort can be {@literal null}.
   * @return
   */
  protected <S extends T> TypedQuery<S> getQuery(Specification<S> spec, Class<S> entityClass, Sort sort) {
    CriteriaBuilder builder = entityManager.getCriteriaBuilder();
    CriteriaQuery<S> query = builder.createQuery(entityClass);

    Root<S> root = applySpecificationToCriteria(spec, entityClass, query);
    query.select(root);

    if (sort != null) {
      query.orderBy(toOrders(sort, root, builder));
    }

    return applyRepositoryMethodMetadata(entityManager.createQuery(query));
  }

  /**
   * Creates a new count query for the given {@link Specification}.
   * 
   * @param spec can be {@literal null}.
   * @return
   */
  protected TypedQuery<Long> getCountQuery(Specification<T> spec) {
    return getCountQuery(spec, getEntityClass());
  }

  /**
   * Creates a new count query for the given {@link Specification}.
   *
   * @param spec can be {@literal null}.
   * @param domainClass must not be {@literal null}.
   * @return
   */
  protected <S extends T> TypedQuery<Long> getCountQuery(Specification<S> spec, Class<S> domainClass) {
    CriteriaBuilder builder = entityManager.getCriteriaBuilder();
    CriteriaQuery<Long> query = builder.createQuery(Long.class);

    Root<S> root = applySpecificationToCriteria(spec, domainClass, query);

    if (query.isDistinct()) {
      query.select(builder.countDistinct(root));
    } else {
      query.select(builder.count(root));
    }

    // Remove all Orders the Specifications might have applied
    query.orderBy(Collections.<Order>emptyList());

    return entityManager.createQuery(query);
  }

  /**
   * Applies the given {@link Specification} to the given {@link CriteriaQuery}.
   *
   * @param spec can be {@literal null}.
   * @param entityClass must not be {@literal null}.
   * @param query must not be {@literal null}.
   * @return
   */
  private <S, U extends T> Root<U> applySpecificationToCriteria(Specification<U> spec, Class<U> entityClass, CriteriaQuery<S> query) {
    Assert.notNull(entityClass, "entityClass 不能为空。");
    Assert.notNull(query, "query 不能为空。");

    Root<U> root = query.from(entityClass);
    if (spec == null) {
      return root;
    }

    CriteriaBuilder builder = entityManager.getCriteriaBuilder();
    Predicate predicate = spec.toPredicate(root, query, builder);

    if (predicate != null) {
      query.where(predicate);
    }

    return root;
  }

  private <S> TypedQuery<S> applyRepositoryMethodMetadata(TypedQuery<S> query) {
    if (metadata == null) {
      return query;
    }

    LockModeType type = metadata.getLockModeType();
    TypedQuery<S> toReturn = type == null ? query : query.setLockMode(type);

    applyQueryHints(toReturn);

    return toReturn;
  }

  private void applyQueryHints(Query query) {
    for (Entry<String, Object> hint : getQueryHints().entrySet()) {
      query.setHint(hint.getKey(), hint.getValue());
    }
  }

  /**
   * Executes a count query and transparently sums up all values returned.
   * 
   * @param query must not be {@literal null}.
   * @return
   */
  private static Long executeCountQuery(TypedQuery<Long> query) {
    Assert.notNull(query, "query 不能为空。");

    List<Long> totals = query.getResultList();
    Long total = 0L;

    for (Long element : totals) {
      total += element == null ? 0 : element;
    }

    return total;
  }

  /**
   * Specification that gives access to the {@link Parameter} instance used to
   * bind the ids for {@link SimpleJpaRepository#findAll(Iterable)}. Workaround
   * for OpenJPA not binding collections to in-clauses correctly when using
   * by-name binding.
   * 
   * @see <a href="https://issues.apache.org/jira/browse/OPENJPA-2018?focusedCommentId=13924055">OPENJPA-2018</a>
   */
  @SuppressWarnings("rawtypes")
  private static final class ByIdsSpecification<T> implements Specification<T> {
    private final JpaEntityInformation<T, ?> entityInformation;
    ParameterExpression<Iterable> parameter;

    public ByIdsSpecification(JpaEntityInformation<T, ?> entityInformation) {
      this.entityInformation = entityInformation;
    }
    
    public Predicate toPredicate(Root<T> root, CriteriaQuery<?> query, CriteriaBuilder cb) {
      Path<?> path = root.get(entityInformation.getIdAttribute());
      parameter = cb.parameter(Iterable.class);
      return path.in(parameter);
    }
  }

  /**
   * {@link Specification} that gives access to the {@link Predicate} instance
   * representing the values contained in the {@link Example}.
   *
   * @param <T>
   */
  private static class ExampleSpecification<T> implements Specification<T> {
    private final Example<T> example;

    /**
     * Creates new {@link ExampleSpecification}.
     *
     * @param example
     */
    public ExampleSpecification(Example<T> example) {
      Assert.notNull(example, "example 不能为空。");
      this.example = example;
    }
    
    @Override
    public Predicate toPredicate(Root<T> root, CriteriaQuery<?> query, CriteriaBuilder cb) {
      return QueryByExamplePredicateBuilder.getPredicate(root, cb, example);
    }
  }
}
